package core

import ( gctx "context"; "kumachan/standalone/ctn" )


type Observable func(DataPublisher)

type DataPublisher struct {
	eventloop  *EventLoop
	context    *context
	observer   *observer
}
type DataSubscriber struct {
	Values     chan <- Object
	Error      chan <- error
	Terminate  chan <- bool
}
type observer struct {
	value     func(Object)
	error     func(error)
	complete  func()
}
func run(run Observable, eventloop *EventLoop, ctx *context, ob *observer) {
	if ctx.isDisposed() {
		return
	}
	var terminated = false
	var terminate = func() { terminated = true }
	var active = func() bool { return !(terminated || ctx.isDisposed()) }
	var proxy = &observer {
		value:    func(v Object) { if active() { ob.value(v) } },
		error:    func(e error)  { if active() { terminate(); ob.error(e) } },
		complete: func()         { if active() { terminate(); ob.complete() } },
	}
	run(DataPublisher {
		eventloop: eventloop,
		context:   ctx,
		observer:  proxy,
	})
}
func Run(o Observable, eventloop *EventLoop, sub DataSubscriber) {
	var V, E, T = sub.Values, sub.Error, sub.Terminate
	eventloop.asyncAddTask_MACRO(func() {
		var ob = &observer {
			value: func(v Object) {
				if V != nil { V <- v }
			},
			error: func(e error) {
				if E != nil { E <- e; close(E) }
				if T != nil { T <- false }
			},
			complete: func() {
				if V != nil { close(V) }
				if T != nil { T <- true }
			},
		}
		run(o, eventloop, nil, ob)
	})
}
func (pub DataPublisher) useInheritedContext() (*context, *observer) {
	return pub.context, pub.observer
}
func (pub DataPublisher) useNewChildContext() (*context, disposeFunc, *observer) {
	var ctx, dispose = pub.context.createChild()
	return ctx, dispose, pub.observer
}
func (pub DataPublisher) run(o Observable, ctx *context, ob *observer) {
	run(o, pub.eventloop, ctx, ob)
}
func (pub DataPublisher) addTimer(ms int, n int, ctx *context, k func()) {
	pub.eventloop.addTimer(ms, n, ctx, k)
}
func (pub DataPublisher) DoSync(k func()(Object,error)) {
	var v, e = k()
	if e == nil {
		pub.observer.value(v)
		pub.observer.complete()
	} else {
		pub.observer.error(e)
	}
}
func (pub DataPublisher) DoSyncWithYield(k func(yield func(Object))(error)) {
	var e = k(pub.observer.value)
	if e == nil {
		pub.observer.complete()
	} else {
		pub.observer.error(e)
	}
}
func (pub DataPublisher) asyncContext() gctx.Context {
	return pub.context.goContext()
}
func (pub DataPublisher) asyncSendValue(v Object) {
	var ob = pub.observer
	pub.eventloop.asyncAddTask_micro(func() {
		ob.value(v)
	})
}
func (pub DataPublisher) asyncSendError(e error) {
	var ob = pub.observer
	pub.eventloop.asyncAddTask_micro(func() {
		ob.error(e)
	})
}
func (pub DataPublisher) asyncSendComplete() {
	var ob = pub.observer
	pub.eventloop.asyncAddTask_micro(func() {
		ob.complete()
	})
}
func (pub DataPublisher) Go(k func(ctx gctx.Context)(Object,error)) {
	go func() {
		var v, e = k(pub.asyncContext())
		if e == nil {
			pub.asyncSendValue(v)
			pub.asyncSendComplete()
		} else {
			pub.asyncSendError(e)
		}
	}()
}
func (pub DataPublisher) GoWithYield(k func(ctx gctx.Context, yield func(Object))(error)) {
	go func() {
		var e = k(pub.asyncContext(), pub.asyncSendValue)
		if e == nil {
			pub.asyncSendComplete()
		} else {
			pub.asyncSendError(e)
		}
	}()
}

type context struct {
	parent     *context
	children   map[*context] struct{}
	disposed   bool
	cleaners   [] func()
	g_context  gctx.Context
	g_dispose  func()
}
type disposeFunc func()
func (ctx *context) createChild() (*context, disposeFunc) {
	var g_ctx, g_dispose = gctx.WithCancel(gctx.Background())
	var parent = ctx
	var child = &context {
		parent:    parent,
		children:  make(map[*context] struct{}),
		disposed:  false,
		cleaners:  make([] func(), 0),
		g_context: g_ctx,
		g_dispose: g_dispose,
	}
	if parent != nil {
		if parent.disposed { panic("something went wrong") }
		parent.children[child] = struct{}{}
	}
	return child, child.__dispose
}
func (ctx *context) __dispose() {
	var parent = ctx.parent
	var child = ctx
	if !(child.disposed) {
		for child_child := range child.children {
			child_child.__dispose()
		}
		if parent != nil {
			delete(parent.children, child)
			child.parent = nil
		}
		for len(child.cleaners) > 0 {
			var l = len(child.cleaners)
			var last = (l - 1)
			child.cleaners[last]()
			child.cleaners[last] = nil
			child.cleaners = child.cleaners[:last]
		}
		child.g_dispose()
	}
}
// registerCleaner registers release operations of persistent resources
func (ctx *context) registerCleaner(h func()) {
	if ctx != nil {
		if !(ctx.disposed) {
			ctx.cleaners = append(ctx.cleaners, h)
		}
	}
}
func (ctx *context) isDisposed() bool {
	return (ctx != nil && ctx.disposed)
}
func (ctx *context) goContext() gctx.Context {
	if ctx != nil {
		return ctx.g_context
	} else {
		return gctx.Background()
	}
}
type CancelError struct {}
func (CancelError) Error() string { return "cancelled" }
func WithCancelSignal(sig Observable, o Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		pub.run(o, ctx, &observer {
			value: ob.value,
			error: func(e error) {
				dispose()
				ob.error(e)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
		pub.run(sig, ctx, &observer {
			value: func(_ Object) {
				dispose()
				ob.error(CancelError {})
			},
			error: func(e error) {
				dispose()
				ob.error(e)
			},
			complete: func() {
				dispose()
				ob.error(CancelError {})
			},
		})
	})
}
func WithCancelTimeout(ms int, o Observable) Observable {
	return WithCancelSignal(SetTimeout(ms), o)
}
func CompleteOnEmit(o Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				dispose()
				ob.complete()
			},
			error: func(e error) {
				dispose()
				ob.error(e)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
	})
}

type Subject struct {
	observerNextId  uint64
	observerList    [] *observer
	observerIndex   map[uint64] uint
	terminated      bool
	maybeError      error
	notifyingFlag   bool
}
func createSubject() *Subject {
	return &Subject {
		observerNextId: 0,
		observerList:   make([] *observer, 0),
		observerIndex:  make(map[uint64] uint),
		terminated:     false,
		maybeError:     nil,
		notifyingFlag:  false,
	}
}
func (b *Subject) Observe() Observable {
	return Observable(func(pub DataPublisher) {
		if b.terminated {
			if b.maybeError != nil {
				var err = b.maybeError
				pub.observer.error(err)
			} else {
				pub.observer.complete()
			}
			return
		}
		var ctx, ob = pub.useInheritedContext()
		var id = b.appendObserver(ob)
		ctx.registerCleaner(func() { b.deleteObserver(id) })
	})
}
func (b *Subject) Value(v Object) Observable {
	return Observable(func(pub DataPublisher) {
		pub.DoSync(func() (Object, error) {
			b.value(v)
			return nil, nil
		})
	})
}
func (b *Subject) Error(err error) Observable {
	return Observable(func(pub DataPublisher) {
		pub.DoSync(func() (Object, error) {
			b.error(err)
			return nil, nil
		})
	})
}
func (b *Subject) Complete() Observable {
	return Observable(func(pub DataPublisher) {
		pub.DoSync(func() (Object, error) {
			b.complete()
			return nil, nil
		})
	})
}
func (b *Subject) value(v Object) {
	if b.terminated {
		return
	}
	b.iterateObservers(func(ob *observer) {
		ob.value(v)
	})
}
func (b *Subject) error(err error) {
	if b.terminated {
		return
	}
	b.terminated, b.maybeError = true, err
	b.iterateObservers(func(ob *observer) {
		ob.error(err)
	})
}
func (b *Subject) complete() {
	if b.terminated {
		return
	}
	b.terminated, b.maybeError = true, nil
	b.iterateObservers(func(ob *observer) {
		ob.complete()
	})
}
func (b *Subject) iterateObservers(k func(*observer)) {
	if b.notifyingFlag { panic("synchronous feedback") }
	b.notifyingFlag = true
	for _, ob := range b.copyObservers() {
		k(ob)
	}
	b.notifyingFlag = false
}
func (b *Subject) copyObservers() ([] *observer) {
	var dup = make([] *observer, len(b.observerList))
	copy(dup, b.observerList)
	return dup
}
func (b *Subject) appendObserver(ob *observer) uint64 {
	var id = b.observerNextId
	var pos = uint(len(b.observerList))
	b.observerList = append(b.observerList, ob)
	b.observerIndex[id] = pos
	b.observerNextId = (id + 1)
	return id
}
func (b *Subject) deleteObserver(id uint64) {
	var pos, exists = b.observerIndex[id]
	if !(exists) { panic("invalid argument") }
	// update index
	delete(b.observerIndex, id)
	for current, _ := range b.observerIndex {
		if current > id {
			// position left shifted
			b.observerIndex[current] -= 1
		}
	}
	// update queue
	b.observerList[pos] = nil
	var L = uint(len(b.observerList))
	if !(L >= 1) { panic("something went wrong") }
	for i := pos; i < (L-1); i += 1 {
		b.observerList[i] = b.observerList[i + 1]
	}
	b.observerList[L-1] = nil
	b.observerList = b.observerList[:L-1]
}
func Multicast(o Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var bus = createSubject()
		ob.value(Obj(bus.Observe()))
		ob.complete()
		pub.run(o, ctx, &observer {
			value:    bus.value,
			error:    bus.error,
			complete: bus.complete,
		})
	})
}


func Noop() Observable {
	return Observable(func(pub DataPublisher) {
		pub.observer.value(nil)
		pub.observer.complete()
	})
}
func SetTimeout(ms int) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.addTimer(ms, 1, ctx, func() {
			ob.value(nil)
			ob.complete()
		})
	})
}
func SetInterval(ms int, n int) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		if n == 0 {
			ob.complete()
			return
		}
		var current = 0
		pub.addTimer(ms, n, ctx, func() {
			if (n < 0) || (current < n) {
				ob.value(ObjInt(current))
				current += 1
			}
			if (n >= 0) && (current == n) {
				ob.complete()
			}
		})
	})
}
func ObservableSequence(forEach func(func(Observable))) Observable {
	return Observable(func(pub DataPublisher) {
		forEach(func(item Observable) {
			pub.observer.value(Obj(item))
		})
		pub.observer.complete()
	})
}

func (o Observable) Catch(f func(error,Observable)(Observable)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.run(o, ctx, &observer {
			error: func(err error) {
				pub.run(f(err,o), ctx, ob)
			},
			value:    ob.value,
			complete: ob.complete,
		})
	})
}
func (o Observable) Retry(limit int) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var retrial = 0
		var proxy *observer
		proxy = &observer {
			error: func(err error) {
				if retrial == limit {
					ob.error(err)
					return
				}
				retrial++
				pub.run(o, ctx, proxy)
			},
			value:    ob.value,
			complete: ob.complete,
		}
		pub.run(o, ctx, proxy)
	})
}

func (o Observable) DistinctUntilChanged(equal func(Object,Object)(bool)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var previous Object
		var available = false
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				if available {
					if equal(v, previous) {
						return
					}
				}
				previous = v
				available = true
				ob.value(v)
			},
			error:    ob.error,
			complete: ob.complete,
		})
	})
}

func (o Observable) WithLatestFrom(another Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		var attached Object
		var available = false
		pub.run(another, ctx, &observer {
			value: func(v Object) {
				attached = v
				available = true
			},
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {},
		})
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				if available {
					ob.value(ObjPair(v, attached))
				}
			},
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
	})
}
func (o Observable) WithCycle(l List) Observable {
	return Observable(func(pub DataPublisher) {
		if l.Empty() {
			pub.observer.complete()
			return
		}
		var ctx, ob = pub.useInheritedContext()
		var node = l.Head
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				ob.value(ObjPair(v, node.Value))
				node = node.Next
				if node == nil { node = l.Head }
			},
			error:    ob.error,
			complete: ob.complete,
		})
	})
}
func (o Observable) WithIndex() Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var i = 0
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				ob.value(ObjPair(v, ObjInt(i)))
				i += 1
			},
			error:    ob.error,
			complete: ob.complete,
		})
	})
}
func (o Observable) WithTime() Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				ob.value(ObjPair(v, ObjTimeNow()))
			},
			error:    ob.error,
			complete: ob.complete,
		})
	})
}

func (o Observable) DelayRun(ms int) Observable {
	return SetTimeout(ms).Then(o)
}
func (o Observable) DelayValues(ms int) Observable {
	return o.MergeMap(func(v Object) Observable {
		return SetTimeout(ms).MapTo(v)
	})
}
func (o Observable) StartWith(first Object) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		ob.value(first)
		pub.run(o, ctx, ob)
	})
}
func (o Observable) EndWith(last Object) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.run(o, ctx, &observer {
			value: ob.value,
			error: ob.error,
			complete: func() {
				ob.value(last)
				ob.complete()
			},
		})
	})
}
func (o Observable) Throttle(f func(Object)(Observable)) Observable {
	return o.ExhaustMap(func(v Object) Observable {
		return CompleteOnEmit(f(v)).StartWith(v)
	})
}
func (o Observable) Debounce(f func(Object)(Observable)) Observable {
	return o.SwitchMap(func(v Object) Observable {
		return CompleteOnEmit(f(v)).EndWith(v)
	})
}
func (o Observable) ThrottleTime(ms int) Observable {
	return o.Throttle(func(_ Object) Observable {
		return SetTimeout(ms)
	})
}
func (o Observable) DebounceTime(ms int) Observable {
	return o.Debounce(func(_ Object) Observable {
		return SetTimeout(ms)
	})
}

func (o Observable) Take(limit int) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		var i = 0
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				ob.value(v)
				i += 1
				if i == limit {
					dispose()
					ob.complete()
				}
			},
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
	})
}
func (o Observable) TakeLast() Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var last Object
		var available = false
		pub.run(o, ctx, &observer {
			error: ob.error,
			value: func(v Object) {
				last = v
				available = true
			},
			complete: func() {
				if available {
					ob.value(last)
				}
				ob.complete()
			},
		})
	})
}
func (o Observable) TakeWhile(f func(Object)(bool)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				if f(v) {
					ob.value(v)
				} else {
					dispose()
					ob.complete()
				}
			},
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
	})
}
func (o Observable) TakeUntil(stop Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		pub.run(o, ctx, &observer {
			value: ob.value,
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
		pub.run(stop, ctx, &observer{
			value: func(_ Object) {},
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
	})
}

func (o Observable) Count() Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var count = 0
		pub.run(o, ctx, &observer {
			error: ob.error,
			value: func(_ Object) {
				count++
			},
			complete: func() {
				ob.value(ObjInt(count))
				ob.complete()
			},
		})
	})
}
func (o Observable) Collect() Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var buf ListBuilder
		pub.run(o, ctx, &observer {
			error: ob.error,
			value: func(v Object) {
				buf.Append(v)
			},
			complete: func() {
				ob.value(Obj(buf.Collect()))
				ob.complete()
			},
		})
	})
}
func (o Observable) BufferTime(ms int) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		var buf ListBuilder
		var renew = func() Object {
			var l = buf.Collect()
			buf = ListBuilder {}
			return Obj(l)
		}
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				buf.Append(v)
			},
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.value(renew())
				ob.complete()
			},
		})
		pub.run(SetInterval(ms, -1), ctx, &observer {
			value: func(_ Object) {
				ob.value(renew())
			},
			error:    func(_ error) { panic("something went wrong") },
			complete: func() { panic("something went wrong") },
		})
	})
}

func (o Observable) Pairwise() Observable {
	return o.BufferCount(2).Map(QueueToPair)
}
func (o Observable) BufferCount(n int) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var sb = createSlidingBuffer(n)
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				var buf, ok = sb.append(v)
				if ok {
					ob.value(buf)
				}
			},
			error:    ob.error,
			complete: ob.complete,
		})
	})
}
type slidingBuffer struct {
	size  int
	mq    ctn.MutQueue[Object]
}
func createSlidingBuffer(size int) *slidingBuffer {
	if size < 1 {
		size = 1
	}
	return &slidingBuffer {
		size: size,
		mq:   ctn.MakeMutQueue[Object](),
	}
}
func (sb *slidingBuffer) append(v Object) (Object, bool) {
	sb.mq.Append(v)
	var diff = (sb.mq.Size() - sb.size)
	if diff < 0 {
		return nil, false
	} else if diff == 0 {
		var buf = ObjQueue(sb.mq.Queue())
		return buf, true
	} else {
		sb.mq.Shift()
		var buf = ObjQueue(sb.mq.Queue())
		return buf, true
	}
}

func (o Observable) Map(f func(Object)(Object)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.run(o, ctx, &observer {
			value:    func(v Object) { ob.value(f(v)) },
			error:    ob.error,
			complete: ob.complete,
		})
	})
}
func (o Observable) MapTo(v Object) Observable {
	return o.Map(func(_ Object)(Object) { return v })
}
func (o Observable) Filter(f func(Object)(bool)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.run(o, ctx, &observer {
			value:    func(v Object) { if f(v) { ob.value(v) } },
			error:    ob.error,
			complete: ob.complete,
		})
	})
}
func (o Observable) Scan(seed Object, f func(Object,Object)(Object)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var current = seed
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				current = f(current, v)
				ob.value(current)
			},
			error:    ob.error,
			complete: ob.complete,
		})
	})
}

func CombineLatest(all ...Observable) Observable {
	return Observable(func(pub DataPublisher) {
		if len(all) == 0 {
			pub.observer.complete()
			return
		}
		var ctx, dispose, ob = pub.useNewChildContext()
		var vv = createValueVector(len(all))
		var completed = 0
		for i_, o := range all {
			var index = i_
			pub.run(o, ctx, &observer {
				value: func(v Object) {
					vv.SetItem(index, v)
					if l, ok := vv.GetList(); ok {
						ob.value(l)
					}
				},
				error: func(err error) {
					dispose()
					ob.error(err)
				},
				complete: func() {
					completed++
					if completed == len(all) {
						dispose()
						ob.complete()
					}
				},
			})
		}
	})
}
func (o Observable) CombineLatest(another Observable) Observable {
	return CombineLatest(o, another).Map(ListToPair)
}
type valueVector struct {
	values     [] Object
	available  [] bool
}
func createValueVector(size int) *valueVector {
	return &valueVector {
		values:    make([] Object, size),
		available: make([] bool, size),
	}
}
func (vv *valueVector) SetItem(index int, value Object) {
	vv.values[index] = value
	if vv.available != nil {
		vv.available[index] = true
	}
}
func (vv *valueVector) GetList() (Object, bool) {
	var ok = true
	if vv.available != nil {
		for _, item_ok := range vv.available {
		if !(item_ok) {
			ok = false
		}}
		if ok { vv.available = nil }
	}
	if ok {
		return ObjList(vv.values), true
	} else {
		return nil, false
	}
}

func (o Observable) Await(k func(Object)(Observable)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		var latest Object
		var ok = false
		pub.run(o, ctx, &observer {
			error: ob.error,
			value: func(v Object) {
				latest = v
				ok = true
			},
			complete: func() {
				if ok {
					pub.run(k(latest), ctx, ob)
				} else {
					ob.complete()
				}
			},
		})
	})
}
func (o Observable) Then(k Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, ob = pub.useInheritedContext()
		pub.run(o, ctx, &observer {
			error:    ob.error,
			value:    func(_ Object) {},
			complete: func() { pub.run(k, ctx, ob) },
		})
	})
}
func (o Observable) With(bg Observable) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		pub.run(bg, ctx, &observer {
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			value:    func(_ Object) {},
			complete: func() {},
		})
		pub.run(o, ctx, &observer {
			value: ob.value,
			error: func(err error) {
				dispose()
				ob.error(err)
			},
			complete: func() {
				dispose()
				ob.complete()
			},
		})
	})
}

func Merge(forEach func(func(Observable))) Observable {
	return ObservableSequence(forEach).MergeMap(GetObservable)
}
func (o Observable) MergeMap(f func(Object)(Observable)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		var proxy = createMergeProxy(ob, dispose)
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				var inner = f(v)
				proxy.innerStart()
				pub.run(inner, ctx, &observer {
					value:    proxy.pass,
					error:    proxy.abort,
					complete: proxy.innerExit,
				})
			},
			error:    proxy.abort,
			complete: proxy.outerClose,
		})
	})
}
type mergeProxy struct {
	observer      *observer
	ctxDispose    disposeFunc
	innerRunning  uint
	outerClosed   bool
}
func createMergeProxy(ob *observer, dispose disposeFunc) *mergeProxy {
	return &mergeProxy {
		observer:     ob,
		ctxDispose:   dispose,
		innerRunning: 0,
		outerClosed:  false,
	}
}
func (p *mergeProxy) pass(x Object) {
	p.observer.value(x)
}
func (p *mergeProxy) abort(e error) {
	p.observer.error(e)
	p.ctxDispose()
}
func (p *mergeProxy) innerStart() {
	p.innerRunning++
}
func (p *mergeProxy) innerExit() {
	if p.innerRunning == 0 { panic("something went wrong") }
	p.innerRunning--
	if p.innerRunning == 0 && p.outerClosed {
		p.observer.complete()
		p.ctxDispose()
	}
}
func (p *mergeProxy) outerClose() {
	p.outerClosed = true
	if p.innerRunning == 0 {
		p.observer.complete()
		p.ctxDispose()
	}
}

func Concat(forEach func(func(Observable))) Observable {
	return ObservableSequence(forEach).ConcatMap(GetObservable)
}
func Concurrent(limit int, forEach func(func(Observable))) Observable {
	return ObservableSequence(forEach).ConcurrentMap(limit, GetObservable)
}
func (o Observable) ConcatMap(f func(Object)(Observable)) Observable {
	return o.ConcurrentMap(1, f)
}
func (o Observable) ConcurrentMap(limit int, f func(Object)(Observable)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		var proxy = createMergeProxy(ob, dispose)
		var queue = createRunQueue(limit, pub, ctx, &observer {
			value:    proxy.pass,
			error:    proxy.abort,
			complete: proxy.innerExit,
		})
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				var inner = f(v)
				proxy.innerStart()
				queue.push(inner)
			},
			error:    proxy.abort,
			complete: proxy.outerClose,
		})
	})
}
func ForkJoin(limit int, all ...Observable) Observable {
	return Observable(func(pub DataPublisher) {
		if len(all) == 0 {
			pub.observer.complete()
			return
		}
		var concurrent = Concurrent(limit, func(yield func(Observable)) {
			for i_, o := range all {
				var index = i_
				yield(o.Map(func(obj Object) Object {
					return ObjPair(obj, ObjInt(index))
				}))
			}
		})
		var ctx, ob = pub.useInheritedContext()
		var vv = createValueVector(len(all))
		pub.run(concurrent, ctx, &observer {
			error: ob.error,
			value: func(obj Object) {
				var v, i_ = GetPair(obj)
				var index = GetInt(i_)
				vv.SetItem(index, v)
			},
			complete: func() {
				if l, ok := vv.GetList(); ok {
					ob.value(l)
				}
				ob.complete()
			},
		})
	})
}
func (o Observable) ForkJoin(limit int, another Observable) Observable {
	return ForkJoin(limit, o, another).Map(ListToPair)
}
type runQueue struct {
	waiting    [] Observable
	running    int
	limit      int
	publisher  DataPublisher
	context    *context
	observer   *observer
}
func createRunQueue(limit int, pub DataPublisher, ctx *context, ob *observer) *runQueue {
	if limit < 1 {
		limit = 1
	}
	var q runQueue
	var proxy = &observer {
		value: ob.value,
		error: ob.error,
		complete: func() {
			ob.complete()
			q.running--
			if len(q.waiting) > 0 {
				var next = q.waiting[0]
				q.waiting[0] = nil
				q.waiting = q.waiting[1:]
				q.publisher.run(next, q.context, q.observer)
			}
		},
	}
	q = runQueue {
		waiting:   make([] Observable, 0),
		running:   0,
		limit:     limit,
		publisher: pub,
		context:   ctx,
		observer:  proxy,
	}
	return &q
}
func (q *runQueue) push(o Observable) {
	if q.running < q.limit {
		q.running++
		q.publisher.run(o, q.context, q.observer)
	} else {
		q.waiting = append(q.waiting, o)
	}
}

func (o Observable) SwitchMap(f func(Object)(Observable)) Observable {
	return Observable(func(pub DataPublisher) {
		var disposeInner disposeFunc
		var wrap = func(o Observable) Observable {
			return Observable(func(pub DataPublisher) {
				if disposeInner != nil {
					disposeInner()
				}
				var ctx, dispose, ob = pub.useNewChildContext()
				disposeInner = func() {
					dispose()
					ob.complete()
				}
				pub.run(o, ctx, ob)
			})
		}
		var ctx, dispose, ob = pub.useNewChildContext()
		var proxy = createMergeProxy(ob, dispose)
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				var inner = f(v)
				proxy.innerStart()
				pub.run(wrap(inner), ctx, &observer {
					value:    proxy.pass,
					error:    proxy.abort,
					complete: proxy.innerExit,
				})
			},
			error:    proxy.abort,
			complete: proxy.outerClose,
		})
	})
}
func (o Observable) ExhaustMap(f func(Object)(Observable)) Observable {
	return Observable(func(pub DataPublisher) {
		var ctx, dispose, ob = pub.useNewChildContext()
		var proxy = createMergeProxy(ob, dispose)
		pub.run(o, ctx, &observer {
			value: func(v Object) {
				if proxy.innerRunning == 0 {
					var inner = f(v)
					proxy.innerStart()
					pub.run(inner, ctx, &observer {
						value:    proxy.pass,
						error:    proxy.abort,
						complete: proxy.innerExit,
					})
				}
			},
			error:    proxy.abort,
			complete: proxy.outerClose,
		})
	})
}


